package com.mango.ch07;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;

import com.mango.HDFSTools.HDFSAPI;

import scala.Tuple2;
import scala.Tuple3;

/**
 * 本demo使用的输入输出路径及资源文件都是共享的mapreduce框架下的MBA_Job的
 * 
 * @author Mango
 *
 */
@SuppressWarnings("unchecked")
public class Spark_MBA_Job {
	static Path inputPath = new Path("/MBA_Job/input");// input
	static Path outPath = new Path("/MBA_Job/output");

	@SuppressWarnings("serial")
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		// 创建输入文件夹
		Configuration conf = new Configuration();
		try {
			HDFSAPI hdfs = new HDFSAPI(conf);
			// hdfs.createDirectory(transactions);
			// 先删除已经有的输出文件夹
			hdfs.delDirectory(outPath);
			// hdfs.orpOver();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.out.println("----------文件操作失败");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		SparkConf sconf = new SparkConf();
		sconf.setAppName("Spark_MBAJob");
		// 建立一个快速串化器
		sconf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
		// 现在默认为32mb的缓冲区而不是64mb
		sconf.set("spark.Kryoserializer.buffer.mb", "32");
		JavaSparkContext jsc = new JavaSparkContext(sconf);
		JavaRDD<String> transcations = jsc.textFile(inputPath.toString(), 1);
		transcations.saveAsTextFile(outPath.toString() + "/1");
		// 生成频繁模式 生成每一行数据的所有模式 如【a,b,c】生成[a],[b],[c],[a,b],[a,c],[b,c][a,b,c]
		JavaPairRDD<List<String>, Integer> patterns = transcations
				.flatMapToPair(new PairFlatMapFunction<String, List<String>, Integer>() {

					@Override
					public Iterator<Tuple2<List<String>, Integer>> call(String arg0) throws Exception {
						// TODO Auto-generated method stub
						List<String> list = toList(arg0);// 将单行数据转换为，业务所需的键值对形式即（（a，b），1）形式
						List<List<String>> combinations = Combination.findSortedCombinatinos(list);// 生成该单行数据的所有模式（即频繁模式）
						List<Tuple2<List<String>, Integer>> result = new ArrayList<>();// 创建一个接受结果的tuple2（该tuple2
																						// 里边存的是键值对）类型的列表集合
						for (List<String> comList : combinations) {
							// 将所有模式便利存入结果集合返回给RDD
							if (comList.size() > 0) {
								result.add(new Tuple2<List<String>, Integer>(comList, 1));
							}
						}
						return result.iterator();
					}

				});
		patterns.saveAsTextFile(outPath.toString() + "/2");
		// 组合规约频繁模式 (reduceByKey 比groupbyKey的效率高很多)
		JavaPairRDD<List<String>, Integer> combined = patterns.reduceByKey(new Function2<Integer, // 输入i1
				Integer, // 输入i2
				Integer// i1+i2的结果
		>() {

			@Override
			public Integer call(Integer arg0, Integer arg1) throws Exception {
				// TODO Auto-generated method stub
				// 如（a,1),(a,1),(b,1) => (a,2),(b,1)
				return arg0 + arg1;
			}
		});
		// 生成所有子模式 将(a,2),(b,1)格式的pairRDD转换为新的格式的pairRDD
		// [格式为：(list(a,b,...),tuple2(List(a,...), requency))]
		JavaPairRDD<List<String>, Tuple2<List<String>, Integer>> subpatterns = combined.flatMapToPair(
				new PairFlatMapFunction<Tuple2<List<String>, Integer>, List<String>, Tuple2<List<String>, Integer>>() {

					@SuppressWarnings("rawtypes")
					@Override
					public Iterator<Tuple2<List<String>, Tuple2<List<String>, Integer>>> call(
							Tuple2<List<String>, Integer> pattern) throws Exception {
						// [格式为： (list(a,b,...),tuple2(List(a,...), requency))]的形式接受结果集
						List<Tuple2<List<String>, Tuple2<List<String>, Integer>>> result = new ArrayList<>();
						List<String> list = pattern._1;// list(a,b,...)
						Integer requency = pattern._2;// integer requency
						result.add(new Tuple2(list, new Tuple2(null, requency)));
						// 如果list中只有一个元素如list（a) 则他的子模式格式为 ([a],null,2))仅有一个
						if (list.size() == 1)
							return result.iterator();
						// 包含多个商品 循环添加
						/**
						 * （原来的键 ，（null），requency），（子模式,（(原来的键），requency）） 如 ([a, b, d],(null,1)) ，([b,
						 * d],([a, b, d],1)) ，([a, d],([a, b, d],1)) ，([a,b],([a, b, d],1))
						 */
						for (int i = 0; i < list.size(); i++) {
							// 如遍历的是[a, b, d]这个集合，
							// 刚好从下标0开始 遍历，生成一个n-1的新list元素即{[b,d],[a,d],[a,b]} 刚好去除每次遍历下标为i那个元素就可以了
							List<String> sublist = removeOneItem(list, i);
							result.add(new Tuple2(sublist, new Tuple2<>(list, requency)));
						}
						return result.iterator();
					}
				});
		subpatterns.saveAsTextFile(outPath.toString() + "/4");
		// 规约组合子模式（相当于 按key分组，规约元素）
		JavaPairRDD<List<String>, Iterable<Tuple2<List<String>, Integer>>> rules = subpatterns.groupByKey();
		rules.saveAsTextFile(outPath.toString() + "/5");
		// 生成关联规则------------- map（） 是一个变换操作，从一个旧的RDD生成一个新的RDD
		JavaRDD<List<Tuple3<List<String>, List<String>, Double>>> assocRules = rules
				.map(new Function<Tuple2<List<String>, Iterable<Tuple2<List<String>, Integer>>>, // 输入
						List<Tuple3<List<String>, List<String>, Double>>>()// 输出
		{
					@Override
					public List<Tuple3<List<String>, List<String>, Double>> call(
							Tuple2<List<String>, Iterable<Tuple2<List<String>, Integer>>> in) throws Exception {
						// TODO Auto-generated method stub
						// 传入的数据格式是 ([a, b],[(null,2), ([a, b, d],1), ([a, b, c],1)])
						List<Tuple3<List<String>, List<String>, Double>> result = new ArrayList<>();
						List<String> fromList = in._1;// key=[a, b]
						Iterable<Tuple2<List<String>, Integer>> to = in._2; // values=[(null,2), ([a, b, d],1), ([a, b,
																			// c],1)]
						List<Tuple2<List<String>, Integer>> toList = new ArrayList<>();
						Tuple2<List<String>, Integer> fromCount = null;
						for (Tuple2<List<String>, Integer> t2 : to) {
							if (t2._1 == null)
								// 找到from对象 即X->Y 中的X对象
								fromCount = t2;// fromCount=(null,2) 记录了基词出现总的频度
							else
								toList.add(t2);// 【([a, b, d],1), ([a, b, c],1)】Y对象list
						}
						// 现在我们得到了生成关联规则所需的对象
						// “from list " "fromCount" "toList" X——>Y Confidence=fromCount
						if (toList.isEmpty())
							// *即没有关联对象
							// 没有生成输出，不过由于spark不接受null对象，我们将模拟一个null对象
							return result;
						// 创建关联规则
						for (Tuple2<List<String>, Integer> t2 : toList) {
							// 如 t2 为([a, b, d],1)
							// fromCount为(null,2)
							// 所以置信度为（x->y 算法为 y出现的频度/x出现的频度)1/2=0.5
							double confidence = (double) t2._2 / (double) fromCount._2;
							List<String> t2List = new ArrayList<>(t2._1);// 如t2._1为 [a, b, d]
							t2List.removeAll(fromList);// 所以t2List为[a, b, d] 移除fromList[a, b] 则剩下[d]
							// 所以输出的格式为 [([a, b],[d],0.5), ([a, b],[c],0.5)]
							result.add(new Tuple3<List<String>, List<String>, Double>(fromList, t2List, confidence));
						}
						return result;
					}

				});
		assocRules.saveAsTextFile(outPath.toString() + "/6");
		jsc.close();
		System.exit(0);
	}

	/**
	 * 将一行字符串切割后转换为一个list
	 * 
	 * @param transaction
	 * @return
	 */
	public static List<String> toList(String transaction) {
		String[] items = transaction.trim().split(",");
		List<String> list = new ArrayList<>();
		for (String string : items) {
			list.add(string);
		}
		return list;
	}

	/**
	 * 移除list中下标为i的元素
	 * 
	 * @param list
	 * @param i
	 * @return
	 */
	public static List<String> removeOneItem(List<String> list, int i) {
		if ((list == null) || (list.isEmpty())) {
			return list;
		}
		if ((i < 0) || i > list.size()) {
			return list;
		}
		List<String> cloned = new ArrayList<>(list);
		cloned.remove(i);
		return cloned;
	}

}
